…ams and runtime walkthrough

…treaming CDC

Address the second sub-case from viirya's review on apache#55636. The previous
contract change covered the same-commit-split-across-micro-batches case via
"all rows of a single commit must appear in the same micro-batch", but missed
the case where two DIFFERENT commits with the same `_commit_timestamp` arrive
in different micro-batches.

Spark's late-event filter and state-eviction predicate both use
`LessThanOrEqual` (`statefulOperators.scala:633-651`), so once a micro-batch
observes max event time T and advances the watermark to T, any later row at
exactly `_commit_timestamp = T` is silently dropped as late. The
atomic-microbatch contract alone doesn't rule this out for distinct commits.

Adds a second contract requirement: distinct `_commit_version` values must
have distinct `_commit_timestamp` values when streaming post-processing is
enabled. Atomic-commit CDC connectors that derive `_commit_timestamp` from
wall-clock time at commit time (Delta, Iceberg) naturally satisfy this.

Doc-only change; no code modifications. The existing tests already exercise
the supported cases; the unsupported case 2 is by definition a connector
contract violation, so we don't add a test for it.

…treaming row-level rewrite

Address @zikangh's review on apache#55637 -- the streaming row-level rewrite should
enforce non-NULL _commit_timestamp, mirroring the runtime guard in
CdcNetChangesStatefulProcessor.

A NULL _commit_timestamp on a streaming read is a connector contract
violation that would silently stall the row's group: the downstream
streaming Aggregate uses _commit_timestamp as an event-time watermark
column AND a grouping key, and Spark's eviction predicate is
LessThanOrEqual(eventTime, watermark) -- a NULL group key never
satisfies that, so the group sits in state until end of stream
producing no output and no error.

Add a Filter at the top of the streaming row-level rewrite that raises
CHANGELOG_CONTRACT_VIOLATION.NULL_COMMIT_TIMESTAMP via the same
RaiseError pattern used for the multiple-changes-per-row-version guard
in the batch path. Also adds the new error class to
error-conditions.json.

Tests:
 - Plan-shape tests: assert the guard Filter is present and sits
   directly above the streaming relation (so it runs before any
   downstream operator sees the NULL).
 - End-to-end test: feeding a row with a NULL _commit_timestamp
   surfaces CHANGELOG_CONTRACT_VIOLATION.NULL_COMMIT_TIMESTAMP at the
   streaming query level rather than producing no output.
 - Existing carry-over / update-detection plan-shape tests updated for
   the extra guard Filter (was 1 -> now 2 Filters in carry-over and
   combined paths; was 0 -> now 1 in update-detection-only).

Also refreshed the addStreamingRowLevelPostProcessing Scaladoc to add a
step 0 (the guard) and step 7 (the watermark-metadata strip), keeping
the per-operator detail aligned with the rewrite's actual shape.

Doc-only side effect: scalafmt reflowed the watermark-metadata bullet
in DataStreamReader.changes() Scaladoc (no semantic change).

Implements `deduplicationMode = netChanges` for DSv2 CDC streaming reads,
completing the streaming post-processing surface. The previous PR
(SPARK-56686) added carry-over removal and update detection for streaming
but kept netChanges batch-only.

The batch path uses a Catalyst Window partitioned by `rowId` and ordered
by `(_commit_version, change_type_rank)` to find the first/last events
per row identity, then applies the SPIP collapse matrix on
`(existedBefore, existsAfter)`. Window is rejected on streaming children.

This PR adds a streaming-friendly equivalent by delegating per-row-identity
state management to a new CdcNetChangesStatefulProcessor driven by
TransformWithState. The processor stores the first event ever observed and
the most-recent event observed for each rowId, and arms an event-time timer
on the latest `_commit_timestamp` for the key. When the global watermark
advances past the timer, handleExpiredTimer evaluates the SPIP matrix and
emits 0, 1, or 2 output rows -- identical semantics to the batch path.

Plan shape (streaming netChanges):
  EventTimeWatermark(_commit_timestamp, 0s)
    -> Project (alias rowId expressions to flat helper columns)
    -> TransformWithState (grouping=rowId helpers, EventTime mode, Append)
    -> SerializeFromObject
    -> Project (drop rowId helper columns)

When carry-over removal / update detection are also requested, the
row-level rewrite is applied first; the netChanges TransformWithState then
sits on top, sharing the same EventTimeWatermark (it's reused rather than
stacked, which would be rejected by the multi-watermark check).

Documented limitation: row identities only touched in the latest observed
commit are held back until a later commit (with strictly greater
`_commit_timestamp`) advances the watermark past them, or the source
terminates. End-of-input flushes all timers, so bounded streams produce
output equivalent to the corresponding batch read.

Removes the now-obsolete error class
INVALID_CDC_OPTION.STREAMING_NET_CHANGES_NOT_SUPPORTED that was introduced
by the previous PR; updates the DataStreamReader.changes() and
Changelog.java Scaladoc to describe the new behavior and the latest-commit
limitation. Tests in ChangelogResolutionSuite,
ResolveChangelogTablePostProcessingSuite, and the new plan-shape suite
flip from "expect throw" to "expect TransformWithState in plan", and
ChangelogEndToEndSuite gains two streaming netChanges end-to-end tests
covering the SPIP matrix's cancel and persist-via-update cases.

…d test fixes

- Tighten Changelog.java to require LONG for _commit_version (post-processing
  compares versions as primitive longs); update the NULL _commit_timestamp
  clause to point to the new error class.
- CdcNetChangesStatefulProcessor: replace the internalError-with-string-prefix
  pattern with structured CHANGELOG_CONTRACT_VIOLATION.UNEXPECTED_CHANGE_TYPE
  (matches batch path) and add a NULL _commit_timestamp guard raising
  CHANGELOG_CONTRACT_VIOLATION.NULL_COMMIT_TIMESTAMP rather than NPE'ing on
  getTime().
- Hoist `inputSchema.fieldIndex("_change_type")` out of `relabel` to avoid a
  linear scan per emitted row.
- Expand `addStreamingNetChangeComputation` Scaladoc Step 1 to describe the
  watermark reuse for combined row-level + netChanges paths.
- ResolveChangelogTableStreamingPostProcessingSuite: assert the
  TransformWithState grouping attributes (so a regression that grouped by the
  wrong attributes can't pass the size check alone).
- ChangelogEndToEndSuite "labels persisting rows as updates": rewrite the
  fixture so the row identity actually exercises (existedBefore=true,
  existsAfter=true) and the (true,true) cell of the SPIP matrix is asserted to
  emit update_preimage + update_postimage; the previous fixture started with
  an INSERT so it fell into the (false,true) cell and emitted INSERT.
- DataStreamReader.changes(): broaden the post-processing "Two implications
  follow" intro from row-level-only to all post-processing modes so the
  netChanges path is covered.

Co-authored-by: Isaac

…rowId

- ChangelogEndToEndSuite: 3 new e2e tests covering missing SPIP matrix cells
  ((true, false) DELETE, (true, true) without computeUpdates), and combined
  netChanges + carry-over removal post-processing path.
- ResolveChangelogTableStreamingPostProcessingSuite: assert grouping
  attribute on netChanges-alone test; add composite rowId plan-shape test.
- Remove stale "deduplicationMode=netChanges is rejected on streaming" test
  (netChanges is now supported on streaming) and the now-unused
  AnalysisException import.

Co-authored-by: Isaac